package com.sxpi.order.config.mq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import java.util.HashMap;
import java.util.Map;

/**
 * @author 飞哥
 * @Title: 学相伴出品
 * @Description: 飞哥B站地址：https://space.bilibili.com/490711252
 * 记得关注和三连哦！
 * @Description: 我们有一个学习网站：https://www.kuangstudy.com
 * @date 2022-09-10$ 14:16$
 */
@Configuration
@Slf4j
public class RabbitMqConfiguration {

    /**
     * 自定设置rabbitmq的工厂实例
     */
    @Autowired
    private CachingConnectionFactory cachingConnectionFactory;

    @Autowired
    private Environment evn;

    /**
     * 创建死信队列
     * @return
     */
    @Bean
    public Queue orderDeadQueue(){
        // 创建一个映射Map。用来指定死信队列的参数信息
        Map<String,Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange",evn.getProperty("mq.order.dead.exchange.name"));
        args.put("x-dead-letter-routing-key",evn.getProperty("mq.order.dead.routing.key.name"));
        // 设置ttl的时间(单位是 ms)，这里我只为了方便测试设置10s,未来你一定是根据业务来决定，然后进行更改
        args.put("x-message-ttl",1000 *  60 * 60 * 24 * 7);
        // 创建死信队列返回
        return new Queue(evn.getProperty("mq.order.dead.queue.name"),true,false,false,args);
    }

    /**
     * 面向生成者 ,用于发消息使用
     * @return
     */
    @Bean
    public TopicExchange orderProducerExchange(){
        return new TopicExchange(
                evn.getProperty("mq.producer.order.exchange.name"),true,false);
    }

    /**
     * 面向生成者 ,绑定关系
     * @return
     */
    @Bean
    public Binding orderProducerBinding(){
        return  BindingBuilder.bind(orderDeadQueue())
                .to(orderProducerExchange())
                .with(evn.getProperty("mq.producer.order.routing.key.name"));
    }

    /**
     * 创建真正的队列，面向消费者
     * @return
     */
    @Bean
    public Queue consumerOrderQueue(){
        return new Queue(evn.getProperty("mq.consumer.order.real.queue.name"),true);
    }

    /**
     * 面向消费者
     * @return
     */
    @Bean
    public TopicExchange orderDeadExchange(){
        return new TopicExchange(
                evn.getProperty("mq.order.dead.exchange.name"),true,false);
    }

    /**
     * 面向消费者
     * @return
     */
    @Bean
    public Binding basicOrderDeadBinding(){
        return BindingBuilder.bind(consumerOrderQueue()).to(orderDeadExchange())
                .with(evn.getProperty("mq.order.dead.routing.key.name"));
    }


    /**
     * 自定义配置RabbitMQ发送消息的操作组件
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate(){
        // 设置发送消息会自动确认
        cachingConnectionFactory.setPublisherReturns(true);
        // 设置消息发送后返回确认信息
        cachingConnectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 构造实例对象
        RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);
        // 一种交换机找不到队列的适合，会把消息返回给生产者
        rabbitTemplate.setMandatory(true);
        // 发送消息后，如果发送成功，则输出“发送成功”的反馈信息
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("消息发送成功,correlationData:{} ack:{},cause:{}",correlationData,ack,cause);
            }
        });

        // 发送消息后，如果发送失败，则输出“消息丢失”的反馈信息
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                log.info("消息丢失：exchange:{}.routeingkey:{},route:{},code:{},text:{},message:{}",
                        returnedMessage.getExchange(),
                        returnedMessage.getRoutingKey(),
                        returnedMessage.getReplyCode(),
                        returnedMessage.getReplyText(),
                        returnedMessage.getMessage()
                );
            }
        });

        return rabbitTemplate;
    }


    /**
     * 单个消消费者的实例配置
     * @return
     */
    @Bean(name="singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainerFactory(){
        // 定义消费监听器所在的容器工厂
        SimpleRabbitListenerContainerFactory
                factory = new SimpleRabbitListenerContainerFactory();
        // 设置容器工厂所用的实例
        factory.setConnectionFactory(cachingConnectionFactory);
        // 设置消息在传输中的格式
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        // 设置消息的应答模式，这里为NONE，标识不需要确认消费
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 设置消费者实例的初始数量，在这里为1个
        factory.setConcurrentConsumers(1);
        // 设置并发消费者的最大数量
        factory.setMaxConcurrentConsumers(1);
        // 设置并发消费者实例中每个实例拉取的消息数量
        factory.setPrefetchCount(1);
        // 返回
        return factory;
    }


    /**
     * 多个消消费者的实例配置
     * @return
     */
    @Bean(name="multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiContainerFactory(){
        // 定义消费监听器所在的容器工厂
        SimpleRabbitListenerContainerFactory
                factory = new SimpleRabbitListenerContainerFactory();
        // 设置容器工厂所用的实例
        factory.setConnectionFactory(cachingConnectionFactory);
        // 设置消息在传输中的格式
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        // 设置消息的应答模式，这里为NONE，标识不需要确认消费
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        // 设置消费者实例的初始数量，在这里为10个
        factory.setConcurrentConsumers(10);
        // 设置并发消费者的最大数量
        factory.setMaxConcurrentConsumers(15);
        // 设置并发消费者实例中每个实例拉取的消息数量
        factory.setPrefetchCount(10);
        // 返回
        return factory;
    }

}
